package ca.tatham.senssim.diffusion;

import java.util.Set;

import ca.tatham.senssim.AbstractNode;
import ca.tatham.senssim.Packet;
import ca.tatham.senssim.RingBuffer;
import ca.tatham.senssim.util.Address;
import ca.tatham.senssim.util.Counter;

public abstract class AbstractDiffusionNode extends AbstractNode
{
  private final static int MAX_CACHED_MESSAGES = 10;
  private final RingBuffer<DataMessage> m_dataCache = new RingBuffer<DataMessage>(
      MAX_CACHED_MESSAGES);
  private final Counter m_interestsSent = new Counter();
  private final Counter m_interestsReceived = new Counter();
  private final Counter m_dataSent = new Counter();
  private final Counter m_dataReceived = new Counter();

  protected AbstractDiffusionNode(Address address, int power)
  {
    super(address, power);
  }

  /**
   * @param message
   * @return true if the message was not already cached
   * @see Set#add(Object)
   */
  protected boolean saveData(DataMessage message)
  {
    return m_dataCache.add(message);
  }

  protected RingBuffer<DataMessage> getDataCache()
  {
    return m_dataCache;
  }

  @Override
  public void processPacket(Packet packet)
  {
    if (packet instanceof InterestMessage)
    {
      m_interestsReceived.increment();
      processInterestMessage((InterestMessage) packet);
    }
    else if (packet instanceof DataMessage)
    {
      m_dataReceived.increment();
      processDataMessage((DataMessage) packet);
    }
  }

  protected abstract void processDataMessage(DataMessage packet);

  protected abstract void processInterestMessage(InterestMessage packet);

  @Override
  public boolean sendPacket(Packet packet)
  {
    if (super.sendPacket(packet))
    {
      if (packet instanceof InterestMessage)
      {
        m_interestsSent.increment();
      }
      else if (packet instanceof DataMessage)
      {
        m_dataSent.increment();
      }
      return true;
    }
    return false;
  }

  public int getInterestsSent()
  {
    return m_interestsSent.get();
  }

  public int getDataSent()
  {
    return m_dataSent.get();
  }

  public int getInterestsReceived()
  {
    return m_interestsReceived.get();
  }

  public int getDataReceived()
  {
    return m_dataReceived.get();
  }

  public abstract boolean isSource();

  public abstract boolean isSink();

}
